package com.pokermind.dse.tcp.client.interactive;

import com.pokermind.common.util.JsonUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 交互式客户端的TCP API客户端
 * 负责与PokerMind服务端进行通信
 */
public class TcpApiClient {

    private final String host;
    private final int port;
    private EventLoopGroup group;
    private Channel channel;
    private final Map<String, CompletableFuture<Map<String, Object>>> pendingRequests = new ConcurrentHashMap<>();
    private final String connectionId;

    public TcpApiClient(String host, int port) {
        this.host = host;
        this.port = port;
        this.connectionId = "interactive_" + UUID.randomUUID().toString().substring(0, 8);
    }

    /**
     * 连接到服务器
     */
    public void connect() throws Exception {
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new ResponseHandler());
                    }
                });

        ChannelFuture future = bootstrap.connect(host, port).sync();
        channel = future.channel();
        System.out.println("✅ 已连接到服务器 " + host + ":" + port);
    }
    
    /**
     * 发送请求并等待响应
     */
    public Map<String, Object> sendRequest(short protocolId, Map<String, Object> payload) throws Exception {
        // 生成请求ID
        String requestId = UUID.randomUUID().toString().replace("-", "");
        CompletableFuture<Map<String, Object>> future = new CompletableFuture<>();
        pendingRequests.put(requestId, future);
        
        // 构造请求包装器（包含rqid, connId, data）
        Map<String, Object> requestWrapper = new LinkedHashMap<>();
        requestWrapper.put("rqid", requestId);
        requestWrapper.put("connId", connectionId);
        requestWrapper.put("data", payload);
        
        // 序列化为JSON
        String jsonRequest = JsonUtil.toJson(requestWrapper);
        byte[] jsonBytes = jsonRequest.getBytes(CharsetUtil.UTF_8);
        
        // 编码并发送（协议格式：Protocol ID (2) + Body Length (4) + JSON Body）
        ByteBuf buffer = Unpooled.buffer();
        buffer.writeShort(protocolId);
        buffer.writeInt(jsonBytes.length);
        buffer.writeBytes(jsonBytes);
        
        channel.writeAndFlush(buffer);
        
        // 等待响应（30秒超时）
        try {
            return future.get(30, TimeUnit.SECONDS);
        } catch (Exception e) {
            pendingRequests.remove(requestId);
            throw new RuntimeException("请求超时或失败: " + e.getMessage(), e);
        }
    }
    
    /**
     * 关闭连接
     */
    public void close() {
        if (channel != null) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
    }
    
    /**
     * 响应处理器
     */
    private class ResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
        private ByteBuf accumulatedBuffer;
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
            try {
                // 累积数据
                if (accumulatedBuffer != null) {
                    ByteBuf combined = ctx.alloc().buffer(accumulatedBuffer.readableBytes() + msg.readableBytes());
                    combined.writeBytes(accumulatedBuffer);
                    combined.writeBytes(msg);
                    accumulatedBuffer.release();
                    accumulatedBuffer = combined;
                } else {
                    accumulatedBuffer = msg.retainedDuplicate();
                }
                
                // 尝试处理完整消息
                handleCompleteMessages();
                
            } catch (Exception e) {
                System.err.println("❌ 处理响应失败: " + e.getMessage());
                e.printStackTrace();
            }
        }
        
        private void handleCompleteMessages() {
            while (accumulatedBuffer != null && accumulatedBuffer.readableBytes() >= 6) {
                // 标记读取位置
                int readerIndex = accumulatedBuffer.readerIndex();
                
                // 读取协议头
                short protocolId = accumulatedBuffer.readShort();
                int bodyLength = accumulatedBuffer.readInt();
                
                // 检查消息体是否完整
                if (accumulatedBuffer.readableBytes() >= bodyLength) {
                    // 读取消息体
                    byte[] bodyBytes = new byte[bodyLength];
                    accumulatedBuffer.readBytes(bodyBytes);
                    String jsonBody = new String(bodyBytes, CharsetUtil.UTF_8);
                    
                    // 解析响应
                    try {
                        @SuppressWarnings("unchecked")
                        Map<String, Object> response = JsonUtil.fromJson(jsonBody, Map.class);
                        
                        // 根据rqid找到对应的Future
                        if (response.containsKey("rqid")) {
                            String requestId = (String) response.get("rqid");
                            CompletableFuture<Map<String, Object>> future = pendingRequests.remove(requestId);
                            if (future != null) {
                                future.complete(response);
                            }
                        }
                    } catch (Exception e) {
                        System.err.println("❌ 解析响应失败: " + e.getMessage());
                    }
                    
                    // 如果还有剩余数据，保留它
                    if (accumulatedBuffer.isReadable()) {
                        ByteBuf remaining = accumulatedBuffer.retainedSlice();
                        accumulatedBuffer.release();
                        accumulatedBuffer = remaining;
                    } else {
                        accumulatedBuffer.release();
                        accumulatedBuffer = null;
                    }
                } else {
                    // 消息不完整，重置读取位置等待更多数据
                    accumulatedBuffer.readerIndex(readerIndex);
                    break;
                }
            }
        }
        
        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            // 清理累积缓冲区
            if (accumulatedBuffer != null) {
                accumulatedBuffer.release();
                accumulatedBuffer = null;
            }
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            System.err.println("❌ 连接异常: " + cause.getMessage());
            ctx.close();
        }
    }
}
